package com.zhu.serialization;

import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.debezium.DebeziumDeserializationSchema;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.source.SourceRecord;
import io.debezium.data.Envelope;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.Struct;


import java.util.List;


/**
 * Flink CDC 2.0 不提供对mysql 5.7以下版本的支持
 * 需要自定义JSON反序列化器具
 * 将数据以JSON格式输出
 */
public class CDCJsonDeserializationSchema implements DebeziumDeserializationSchema<String> {


    /**
     * @param sourceRecord 输入 SourceRecord{sourcePartition={server=mysql_binlog_source},
     *                    sourceOffset={transaction_id=null, ts_sec=1661421973, file=mysql-bin.000032, pos=5107,
     *                    gtids=186ee13a-e180-11ec-b32b-000c29d17377:1-38, row=1, server_id=1, event=2}}
     *                    ConnectRecord{topic='mysql_binlog_source.cdc_test.user_info', kafkaPartition=null, key=Struct{id=1006},
     *                     keySchema=Schema{mysql_binlog_source.cdc_test.user_info.Key:STRUCT},
     *                     value=Struct{before=Struct{id=1006,name=aa,sex=aa},
     *                     after=Struct{id=1006,name=aaa,sex=female},
     *                     source=Struct{version=1.5.2.Final,connector=mysql,name=mysql_binlog_source,ts_ms=1661421973000
     *                     ,db=cdc_test,table=user_info,server_id=1,gtid=186ee13a-e180-11ec-b32b-000c29d17377:39,
     *                     file=mysql-bin.000032,pos=5239,row=0},op=u,ts_ms=1661421974171},
     *                     valueSchema=Schema{mysql_binlog_source.cdc_test.user_info.Envelope:STRUCT}, timestamp=null, headers=ConnectHeaders(headers=)}
     * @param collector    输出 {"db":"",
     *                     "tableName":"",
     *                     "before":"{"id":"1001","name":"zhangsan","sex":"male"}",
     *                     "after":"{"id":"1002","name":"lisi","sex":"female"}",
     *                     "op":"", //操作类型
     *                     }
     * @throws Exception
     */
    @Override
    public void deserialize(SourceRecord sourceRecord, Collector collector) throws Exception {

        //1.输出为json格式
        //1.1创建一个json对象用于封装输出结果
        JSONObject jsonObject = new JSONObject();
       // System.out.println(sourceRecord.value());

        //2.1获取 db & tableName 以.做分隔符取第二第三字段作为库名表名
        String topic = sourceRecord.topic(); //{topic=mysql_binlog_source.gmall-flink-config.table_process
       // System.out.println(topic);
        String[] dbNAndTblN = topic.split("\\.");  //从topic中提取数据库名 和 列名
        jsonObject.put("db", dbNAndTblN[1]);
        jsonObject.put("tableName", dbNAndTblN[2]);

        //2.2获取before数据
        //2.2.1所需数据在struct结构体内 将获取的value转换为struct结构
        //value=Struct{after=Struct{source_table=aa,sink_table=bb,sink_columns=cc,sink_pk=dd,sink_extend=xxx},
        // source=Struct{version=1.5.2.Final,connector=mysql,name=mysql_binlog_source,ts_ms=1679645558017,snapshot=last,
        // db=gmall-flink-config,table=table_process,server_id=0,file=mysql-bin.000029,pos=154,row=0},op=r,ts_ms=1679645558021}
        Struct value = (Struct) sourceRecord.value();
        //嵌套结构体
        Struct before = value.getStruct("before");
        //2.2.3创建json对象封装before对象
        JSONObject beforeJson = new JSONObject();
        if (before != null) {
            //2.2.4获取before结构体的元数据信息
            Schema schema = before.schema();
            //2.2.5获取列名
            List<Field> fields = schema.fields();
            for (Field field : fields) {
                beforeJson.put(field.name(), before.get(field));  //转换成json格式
            }
        }
        jsonObject.put("before", beforeJson);
        //2.3获取after数据
        Struct after = value.getStruct("after");
        JSONObject afterJson = new JSONObject();
        if (after != null) {
            Schema schema = after.schema();
            List<Field> fields = schema.fields();
            for (Field field : fields) {
                afterJson.put(field.name(), after.get(field));
            }
        }
        jsonObject.put("after", afterJson);

        //获取操作类型 READ DELETE UPDATE CREATE
        Envelope.Operation operation = Envelope.operationFor(sourceRecord);
        jsonObject.put("op",operation);

        //timeStamp
        Long timestamp = (Long) value.get("ts_ms");
        jsonObject.put("ts_ms", timestamp.toString());
        System.out.println(jsonObject.toJSONString());



        //输出数据
        collector.collect(jsonObject.toJSONString());
    }

    /**
     * 获取类型
     * @return String
     */
    @Override
    public TypeInformation<String> getProducedType() {
        return BasicTypeInfo.STRING_TYPE_INFO;
    }

}
